package cn.edu.ccnu.nlp;

import java.io.IOException;
import java.sql.*;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
public class DataImport {	 
	
	/**
	 * 获取mysql数据库的连接
	 * @return
	 */
	public Connection getConnection(){
	    String url = "jdbc:mysql://202.114.40.175:3306/corpus?useUnicode=true&characterEncoding=UTF-8";
		String name = "root";
		String psw = "ccnunlp2010@nlp.com";
		Connection conn = null;
		try {
			Class.forName("com.mysql.jdbc.Driver");
			conn = DriverManager.getConnection(url, name, psw);
			System.out.println("连接成功");
		} catch (Exception e) {
			e.printStackTrace();
		}
		return conn;
	}
	
	public static Configuration configuration;
	static {
		configuration = HBaseConfiguration.create();
		//configuration.set("hbase.zookeeper.property.clientPort", "2181");
		//configuration.set("hbase.zookeeper.quorum", "192.168.1.100");
		//configuration.set("hbase.master", "192.168.1.100:600000");
	}
	
	/**
	 * create a hbase table
	 * @param tablename
	 * @param familyname
	 * @throws IOException
	 */
	public void createNews (String tablename) throws IOException {	 
		 HBaseAdmin admin = new HBaseAdmin(configuration);
		 HTableDescriptor tabledescriptor = new HTableDescriptor(Bytes.toBytes("news"));
	
		 tabledescriptor.addFamily(new HColumnDescriptor ("url"));
		 tabledescriptor.addFamily(new HColumnDescriptor ("title"));
		 tabledescriptor.addFamily(new HColumnDescriptor ("author"));
		 tabledescriptor.addFamily(new HColumnDescriptor ("date"));
		 tabledescriptor.addFamily(new HColumnDescriptor ("source"));
		 tabledescriptor.addFamily(new HColumnDescriptor ("html"));
		 tabledescriptor.addFamily(new HColumnDescriptor ("text"));
		 tabledescriptor.addFamily(new HColumnDescriptor ("seg"));
		 
		 admin.createTable(tabledescriptor);
			 
		}
	/**
	 * 创建news_stat表
	 * @param tablename
	 * @throws IOException
	 */
	public void createNewsStat (String tablename) throws IOException {	 
		 HBaseAdmin admin = new HBaseAdmin(configuration);
		 HTableDescriptor tabledescriptor = new HTableDescriptor(Bytes.toBytes("news_stat"));
	
		 tabledescriptor.addFamily(new HColumnDescriptor ("date"));
		 tabledescriptor.addFamily(new HColumnDescriptor ("num"));		 
		 admin.createTable(tabledescriptor);
			 
		}
	/**
	 * 向news表中插入数据
	 * @param tablename
	 * @param url
	 * @param title
	 * @param author
	 * @param date
	 * @param source
	 * @param html
	 * @param text
	 * @param seg
	 * @throws IOException
	 */
	public void insertNews(String tablename,String url,String title,String author,String date,String source,String html,String text,String seg) throws IOException{	 
		HTable table = new HTable(configuration, tablename);
		Put put = new Put(Bytes.toBytes(url));
		put.add(Bytes.toBytes("url"), null,	url.getBytes());
		put.add(Bytes.toBytes("title"), null,title.getBytes());
		put.add(Bytes.toBytes("author"), null,	author.getBytes());
		put.add(Bytes.toBytes("date"), null,date.getBytes());
		put.add(Bytes.toBytes("source"), null,	source.getBytes());
		put.add(Bytes.toBytes("html"), null,html.getBytes());
		put.add(Bytes.toBytes("text"), null,text.getBytes());
		put.add(Bytes.toBytes("seg"), null,seg.getBytes());
		table.put(put);
	}
	
	/**
	 * 向news_stat表中插入数据
	 * @param tablename
	 * @param date
	 * @param num
	 * @throws IOException
	 */
	public void insertNewsStat(String tablename,String date,String num) throws IOException{	 
		HTable table = new HTable(configuration, tablename);
		Put put = new Put(Bytes.toBytes(date));
		put.add(Bytes.toBytes("date"), null,date.getBytes());
		put.add(Bytes.toBytes("num"), null,num.getBytes());
		table.put(put);
	}
	
	/**
	 *将mysql的news_stat表迁移到hbase
	 */
	public  void importNewsStat(){
		Connection conn = null;
		Statement stmt=null;
		ResultSet rs=null;
		String sql = null;
		try 
		{
			conn = getConnection();
			stmt = conn.createStatement();
		    /**
		     * 获取表中总的记录数
		     */
		    sql="SELECT count(*)  FROM news_stat";
		   // 
		    rs = stmt.executeQuery(sql);
		    rs.next();
		    int count = rs.getInt(1);
		    System.out.println("总记录数为："+count);
		    
		    /**
		     * 循环获取数据
		     */
		    int j = 1;
		    for(int i = 0; i <= count/1000;i++){
		    	conn = getConnection();
				stmt = conn.createStatement();
		    	System.out.println("准备导入第"+(i+1)*1000+"条....................");
			    String sql_2="SELECT *  FROM news_stat limit "+i*1000+","+1000;
			    System.out.println(sql_2);
			    rs = stmt.executeQuery(sql_2);
			    while(rs.next())
			    {
			    	System.out.println(rs.getString("date")+":"+rs.getString("num"));
			    	insertNewsStat("news_stat", rs.getString("date"),rs.getString("num"));		    	
			    }
			    System.out.println("第"+(i+1)*1000+"条记录导入成功！");	
			    rs.close();
			    conn.close();
		    }
		}		
		catch (Exception e) 
		{
		  	e.printStackTrace(System.out);
		}
	}
	
	/**
	 *将mysql的news表迁移到hbase
	 *2677193条记录
	 */
	public void importNews(){
		Connection conn = null;
		Statement stmt=null;
		ResultSet rs=null;
		String sql = null;
		try 
		{
			conn = getConnection();
		    stmt = conn.createStatement();
		    /**
		     * 获取表中总的记录数
		     */
		    sql ="SELECT count(*)  FROM news";
		   // 
		    rs = stmt.executeQuery(sql);
		    rs.next();
		    int count = rs.getInt(1);
		    System.out.println("总记录数为："+count);
		    rs.close();
		    conn.close();
		    /**
		     * 循环获取数据
		     */
		    for(int i = 0; i <= count/1000;i++){
		    	conn = getConnection();
			    stmt = conn.createStatement();
		    	System.out.println("准备导入第"+(i+1)*1000+"条....................");
			    String sql_2="SELECT *  FROM news limit "+i*1000+","+1000;
			    System.out.println(sql_2);
			    rs = stmt.executeQuery(sql_2);
			    while(rs.next())
			    {
			    	String author;
			    	if(rs.getString("author") == null)
			    		author = "Null";
			    	else 
			    		author = rs.getString("author");
			    	insertNews("news", rs.getString("url"), rs.getString("title"),
					author, rs.getString("date")+":"+rs.getString("time"), 
					rs.getString("source"), "html", rs.getString("content"), "seg");  	
			    }
			    rs.close();
			    conn.close();
			    System.out.println("第"+(i+1)*1000+"条记录导入成功！");	
		    }
		}		
		catch (Exception e) 
		{
		  	e.printStackTrace(System.out);
		}
		
	}
	
	  public static void main(String[] args) throws IOException
	  {		 
			
		  DataImport di = new DataImport();
//		  di.importData();
//		  DataImpl di = new DataImpl();
//		  di.QueryAll("news");
		  //di.createNewsStat("news_stat");
		  //di.importNewsStat();
	  }	  
}

